package com.spring.component.rabbitmq.listener;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import com.spring.component.rabbitmq.sender.DefaultSender;
import com.spring.component.redis.RedisUtils;
import com.spring.component.utils.HttpClientUtil;
import com.spring.component.utils.ThreadPoolUtil;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * @author liushihua
 * @since 2018年04月04日
 */
@Component
public class OutBoundListener {
  protected final Logger log = LoggerFactory.getLogger(this.getClass());
  @Autowired
  private RedisUtils redisUtils;
  @Autowired
  private ThreadPoolUtil threadPoolUtil;
  @Autowired
  private DefaultSender defaultSender;

  // @RabbitListener注释方法的时候，方法支持多种参数，其中有一个是Channel。通过Channel.basicAck和Channel.basicNack实现
  // @RabbitListener(queues = "MSG_OUTBOUND_APP8075")
  public void process(Message message) {
    String busMsg = new String(message.getBody());
    log.info("接收消息----{}", busMsg);

    log.info("-----------------------线程数：{}", threadPoolUtil.getExecutor().getPoolSize());
    /*
     * if (ThreadPoolUtil.isFull()) { // 如果线程池已满，主线程等待，不消费消息 try { log.info("主线程等待..."); //
     * Thread.sleep(300); ThreadPoolUtil.acquire(); log.info("有子线程处理结束，主线程继续消费消息"); } catch
     * (InterruptedException e) { e.printStackTrace(); } }
     */
    try {
      // log.info("主线程等待...");
      // Thread.sleep(300);
      // threadPoolUtil.acquire();
      // log.info("有子线程处理结束，主线程继续消费消息");

      // rabbitmq自身就支持多线程消费消息
      threadPoolUtil.getExecutor().execute(new Runnable() {
        @Override
        public void run() {
          try {
            log.info("线程{}工作中...", Thread.currentThread().getName());
            // Thread.sleep(300);
            for (int i = 0; i < 11; i++) {
              HttpClientUtil.doGet("http://www.baidu.com");
            }
            log.info("开门计数----{}", redisUtils.generate("count"));
            log.info("线程{}工作完成!!!", Thread.currentThread().getName());
            threadPoolUtil.release();
          } catch (Exception e) {
            e.printStackTrace();
          }
        }
      });
    } catch (Exception e) {
      log.error("异常----", e);
    }
  }

  // @RabbitListener(queues = "DEFAULT_TOPIC_TEST_03", containerFactory = "rlcFactory")
  public void testListener(Message message) {
    log.info("接收消息：{}", message.toString());
    log.info("工作....");
    Channel channel = null;
    try {
      channel = ((CachingConnectionFactory) defaultSender.getConnectionFactory())
          .getRabbitConnectionFactory().newConnection().createChannel();
      // consumer发送ACK，Delivery Tag 用来标识信道中投递的消息。RabbitMQ 推送消息给 Consumer 时，会附带一个 Delivery
      // Tag,false表示当前消息被确认
      // channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    } catch (IOException e) {
      e.printStackTrace();
    } catch (TimeoutException e) {
      e.printStackTrace();
    }
    try {
      Thread.sleep(10000);
      log.info("工作完成");
      throw new NullPointerException();
    } catch (Exception e) {
      log.error("处理消息发生异常，取消确认");
      /*
       * try { // consumer发送NACK，取消确认，false为丢弃该条消息，true消息重新放到队列头,如果DeliveryTag不存，将会抛出异常，关闭channel
       * channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true); } catch
       * (IOException e1) { e1.printStackTrace(); }
       */
    }
  }

  @RabbitListener(queues = "MSG_INBOUND_APP8075", containerFactory = "rlcFactory")
  public void testbasicReject(Message message) {
    log.info("接收消息：{}", message.toString());
    log.info("线程{}工作中...", Thread.currentThread().getName());
    try {
      Thread.sleep(300);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    log.info("开门计数----{}", redisUtils.generate("count"));
    log.info("线程{}工作完成!!!", Thread.currentThread().getName());
  }

  // @RabbitListener(queues = "MSG_INBOUND_APP8075", containerFactory = "rlcFactory")
  public void testInbound(Message message) {
    log.info("接收消息：{}", message.toString());
    Channel channel = null;
    try {
      channel = ((CachingConnectionFactory) defaultSender.getConnectionFactory())
          .getRabbitConnectionFactory().newConnection().createChannel();
    } catch (IOException e) {
      e.printStackTrace();
    } catch (TimeoutException e) {
      e.printStackTrace();
    }
    try {
      Thread.sleep(1000);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    try {
      log.info("手动应答......");
      channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    } catch (IOException e) {
      e.printStackTrace();
    }
  }
}
